# Shell注释要用# 但是复制的时候 -- 没改成# 


#!/bin/bash 
# 作者：李凯  版本号 3.5 
# 功能：ODS→DWD→DWS→ADS全自动增量同步，非交互式，支持n>=2次插入
# 特性：表名全部大写 / T+1模式 / 可配置计划任务 
 
# 获取昨天日期（T+1同步基准）
YESTERDAY=$(date -d "yesterday" +%Y-%m-%d)
# 获取当前系统时间（用于日志时间戳）
CURRENT_TIME=$(date +'%Y-%m-%d %H:%M:%S')


# 日志配置
# 定义日志目录（全链路增量同步日志）
LOG_DIR="/kai_project/logs/full_layer_increment"
# 若目录不存在则创建
mkdir -p ${LOG_DIR}
# 日志文件路径（按日期区分）
LOG_FILE="${LOG_DIR}/sync_${YESTERDAY}.log"
# 初始化日志文件（覆盖写入启动信息）
echo "[${CURRENT_TIME}] 增量插入启动（目标日期：${YESTERDAY}）" > ${LOG_FILE}


# 2. Hive核心配置（动态分区支持）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 执行Hive全局配置ing" >> ${LOG_FILE}
hive -e "
set hive.exec.dynamic.partition=true;  # 开启动态分区功能（允许自动创建分区）
set hive.exec.dynamic.partition.mode=nonstrict;  # 非严格模式（允许所有分区字段动态生成）
set hive.exec.max.dynamic.partitions=1000;  # 最大动态分区数（避免元数据压力）
set hive.exec.max.dynamic.partitions.pernode=100;  # 单节点最大分区数（控制节点负载）
" >> ${LOG_FILE} 2>&1  # 标准输出和错误输出均追加到日志

# 检查Hive配置是否成功
if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] Hive配置执行失败！" >> ${LOG_FILE}
    exit 1
fi


# 3. ODS→DWD 增量同步（仅插入）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 开始ODS→DWD增量同步ing" >> ${LOG_FILE}


# 3.1 客户主表（需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步客户主表（DWD_DIM_CUSTOMER_MASTER）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_DIM_CUSTOMER_MASTER PARTITION (etl_date)
SELECT
    customer_id,
    cust_no,
    -- 姓名脱敏：2字以上保留首尾字符，中间1星；2字及以下保留首字符+1星
    CASE
        WHEN cust_name IS NOT NULL AND length(cust_name) > 2
        THEN concat(substr(cust_name, 1, 1),'*',substr(cust_name,-1))
        ELSE concat(substr(cust_name, 1, 1),'*')
    END AS cust_name,
    -- 客户类型转换
    CASE
        WHEN cust_type = 'IND' THEN '个人'
        WHEN cust_type = 'CORP' THEN '公司'
    END AS cust_type,
    id_type,
    -- 身份证号脱敏：保留前6位+后4位，中间用*填充
    CASE
        WHEN id_no IS NOT NULL AND length(id_no) > 10
        THEN concat(
            substr(id_no, 1, 6),
            repeat('*', length(id_no) - 10),
            substr(id_no, -4)
        )
        ELSE id_no
    END AS id_no,
    gender,
    birth_date,
    nationality,
    residence_country,
    residence_city,
    -- 地址脱敏：保留前4位+后4位，中间3星
    CASE
        WHEN address IS NOT NULL AND length(address) > 8
        THEN concat(
            substr(address, 1, 4),
            '***',
            substr(address, -4)
        )
        ELSE address
    END AS address,
    -- 手机号脱敏：11位手机号保留前3位+后4位，中间4星
    CASE
        WHEN phone IS NOT NULL AND length(phone) = 11
        THEN concat(
            substr(phone, 1, 3),
            '****',
            substr(phone, -4)
        )
        ELSE phone
    END AS phone,
    -- 邮箱脱敏：用户名保留前2位，其余*，保留域名
    CASE
        WHEN email IS NOT NULL AND email LIKE '%@%'
        THEN concat(
            substr(split(email, '@')[0], 1, 2),
            repeat('*', greatest(length(split(email, '@')[0])-2, 0)),
            '@',
            split(email, '@')[1]
        )
        ELSE email
    END AS email,
    occupation,
    industry,
    annual_income,
    source_of_funds,
    purpose_of_account,
    risk_level,
    -- 风险等级来源转换
    case when risk_level_source = 'MANUAL' then '人工'
        when risk_level_source = 'AUTO' then '系统'
    end as risk_level_source,
    risk_assess_date,
    open_date,
    close_date,
    status,
    created_by,
    created_date,
    updated_by,
    updated_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值（昨天日期）
FROM ODS.ODS_CUSTOMER_MASTER
WHERE etl_date = '${YESTERDAY}';  -- 仅同步昨天的增量数据
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 客户主表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.2 交易流水表（需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步交易流水表（DWD_FACT_TRANSACTION_DETAIL）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_FACT_TRANSACTION_DETAIL PARTITION (etl_date)
SELECT
    txn_id,
    txn_seq_no,
    account_id,
    txn_type,
    txn_sub_type,
    txn_channel,
    txn_amount,
    currency,
    txn_date,
    txn_time,
    txn_status,
    core_txn_id,
    ref_no,
    -- 汇款人姓名脱敏：同客户姓名规则
    CASE
        WHEN remitter_name IS NOT NULL AND length(remitter_name) > 2
        THEN concat(substr(remitter_name, 1, 1), '*',substr(remitter_name, -1))
        ELSE concat(substr(remitter_name, 1, 1), '*')
    END AS remitter_name,
    -- 汇款人账号脱敏：保留前4位+后4位，中间*填充
    CASE
        WHEN remitter_acct_no IS NOT NULL AND length(remitter_acct_no) > 8
        THEN concat(
            substr(remitter_acct_no, 1, 4),
            repeat('*', length(remitter_acct_no) - 8),
            substr(remitter_acct_no, -4)
        )
        ELSE remitter_acct_no
    END AS remitter_acct_no,
    -- 收款人姓名脱敏：同客户姓名规则
    CASE
        WHEN beneficiary_name IS NOT NULL AND length(beneficiary_name) > 2
        THEN concat(substr(beneficiary_name, 1, 1), '*', substr(beneficiary_name, -1))
        ELSE concat(substr(beneficiary_name, 1, 1), '*')
    END AS beneficiary_name,
    -- 收款人账号脱敏：同汇款人账号规则
    CASE
        WHEN beneficiary_acct_no IS NOT NULL AND length(beneficiary_acct_no) > 8
        THEN concat(
            substr(beneficiary_acct_no, 1, 4),
            repeat('*', length(beneficiary_acct_no) - 8),
            substr(beneficiary_acct_no, -4)
        )
        ELSE beneficiary_acct_no
    END AS beneficiary_acct_no,
    remitter_country,
    beneficiary_country,
    ip_address,
    device_id,
    geo_location,
    created_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
FROM ODS.ODS_TRANSACTION_DETAIL
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 交易流水表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.3 受益人信息表（需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步受益人信息表（DWD_DIM_UBO_INFO）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_DIM_UBO_INFO PARTITION (etl_date)
SELECT
    ubo_id,
    customer_id,
    -- UBO姓名脱敏：同客户姓名规则
    CASE
        WHEN ubo_name IS NOT NULL AND length(ubo_name) > 2
        THEN concat(substr(ubo_name, 1, 1), '*',substr(ubo_name, -1))
        ELSE concat(substr(ubo_name, 1, 1), '*')
    END AS ubo_name,
    ubo_id_type,
    -- UBO身份证号脱敏：同客户身份证规则
    CASE
        WHEN ubo_id_no IS NOT NULL AND length(ubo_id_no) > 10
        THEN concat(
            substr(ubo_id_no, 1, 6),
            repeat('*', length(ubo_id_no) - 10),
            substr(ubo_id_no, -4)
        )
        ELSE ubo_id_no
    END AS ubo_id_no,
    ubo_share_percent,
    ubo_position,
    ubo_nationality,
    -- UBO居住地脱敏：同地址规则
    CASE
        WHEN ubo_residence IS NOT NULL AND length(ubo_residence) > 8
        THEN concat(
            substr(ubo_residence, 1, 4),
            '***',
            substr(ubo_residence, -4)
        )
        ELSE ubo_residence
    END AS ubo_residence,
    relationship_type,
    effective_date,
    expiry_date,
    status,
    created_by,
    created_date,
    updated_by,
    updated_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
FROM ODS.ODS_UBO_INFO
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 受益人信息表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.4 名单明细表（需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步名单明细表（DWD_DIM_WATCHLIST_DETAIL）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_DIM_WATCHLIST_DETAIL PARTITION (etl_date)
SELECT
    entity_id,
    list_id,
    -- 实体类型转换
    case when entity_type = 'PERSON' then '个人'
        when entity_type = 'ORG' then '组织'
    end as entity_type,
    -- 实体名称脱敏：同客户姓名规则
    CASE
        WHEN name IS NOT NULL AND length(name) > 2
        THEN concat(substr(name, 1, 1), '*',substr(name, -1))
        ELSE concat(substr(name, 1, 1), '*')
    END AS name,
    -- 别名脱敏：同实体名称规则
    CASE
        WHEN alias_name IS NOT NULL AND length(alias_name) > 2
        THEN concat(substr(alias_name, 1, 1), '*', substr(alias_name, -1))
        ELSE concat(substr(alias_name, 1, 1), '*')
    END AS alias_name,
    id_type,
    -- 证件号码脱敏：同身份证规则
    CASE
        WHEN id_no IS NOT NULL AND length(id_no) > 10
        THEN concat(
            substr(id_no, 1, 6),
            repeat('*', length(id_no) - 10),
            substr(id_no, -4)
        )
        ELSE id_no
    END AS id_no,
    nationality,
    birth_date,
    place_of_birth,
    -- 地址脱敏：同地址规则
    CASE
        WHEN address IS NOT NULL AND length(address) > 8
        THEN concat(
            substr(address, 1, 4),
            '***',
            substr(address, -4)
        )
        ELSE address
    END AS address,
    listed_date,
    delisted_date,
    reason,
    sanction_type,
    reference_info,
    status,
    created_date,
    updated_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
FROM ODS.ODS_WATCHLIST_DETAIL
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 名单明细表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.5 筛查结果表（需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步筛查结果表（DWD_FACT_SCREENING_RESULT）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_FACT_SCREENING_RESULT PARTITION (etl_date)
SELECT
    result_id,
    -- 筛查类型转换
    case screening_type
        when 'CUSTOMER' THEN '客户开户'
        when 'TRANSACTION' THEN '交易'
        when 'PERIODIC' THEN '定期回溯'
    end as screening_type,
    screening_date,
    screening_time,
    target_type,
    target_id,
    -- 目标名称脱敏：同客户姓名规则
    CASE
        WHEN target_name IS NOT NULL AND length(target_name) > 2
        THEN concat(substr(target_name, 1, 1),'*',substr(target_name, -1))
        ELSE concat(substr(target_name, 1, 1),'*')
    END AS target_name,
    matched_entity_id,
    matched_list_id,
    match_score,
    match_level,
    screening_status,
    reviewed_by,
    reviewed_date,
    comments,
    created_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
FROM ODS.ODS_SCREENING_RESULT
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 筛查结果表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.6 账户主表（无需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步账户主表（DWD_DIM_ACCOUNT_MASTER）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_DIM_ACCOUNT_MASTER PARTITION (etl_date)
select 
    account_id, 
    acct_no, 
    customer_id, 
    acct_type, 
    currency, 
    open_date, 
    close_date, 
    status, 
    current_balance, 
    avg_daily_balance, 
    last_txn_date, 
    channel_open, 
    branch_code, 
    branch_name, 
    created_by, 
    created_date, 
    updated_by, 
    updated_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
from ODS.ODS_ACCOUNT_MASTER
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 账户主表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.7 名单主表（无需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步名单主表（DWD_DIM_WATCHLIST_MASTER）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_DIM_WATCHLIST_MASTER PARTITION (etl_date)
select 
    list_id, 
    list_code, 
    list_name, 
    list_source, 
    list_type, 
    currency, 
    country, 
    valid_from, 
    valid_to, 
    status, 
    description, 
    updated_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
from ODS.ODS_WATCHLIST_MASTER
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 名单主表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.8 预警表（无需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步预警表（DWD_FACT_ALTER）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_FACT_ALTER PARTITION (etl_date)
select 
    alert_id, 
    alert_no, 
    customer_id, 
    account_id, 
    rule_id, 
    rule_name, 
    alert_type, 
    alert_date, 
    alert_time, 
    alert_status, 
    severity_level, 
    score, 
    trigger_amount, 
    trigger_txn_count, 
    description, 
    assigned_to, 
    assigned_date, 
    reviewed_by, 
    reviewed_date, 
    comments, 
    str_no, 
    created_date, 
    updated_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
from ODS.ODS_ALTER
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 预警表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.9 可疑交易报告表（无需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步可疑交易报告表（DWD_FACT_SUSPICIOUS_TXN_REPORT）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_FACT_SUSPICIOUS_TXN_REPORT PARTITION (etl_date)
select 
    str_id, 
    str_no, 
    customer_id, 
    report_date, 
    submit_date, 
    report_status, 
    report_type, 
    case_category, 
    suspicion_basis, 
    total_amount, 
    currency, 
    first_txn_date, 
    last_txn_date, 
    related_alert_ids, 
    filed_by, 
    filed_date, 
    approved_by, 
    approved_date, 
    regulator_ref_no, 
    comments,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
from ODS.ODS_SUSPICIOUS_TXN_REPORT
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 可疑交易报告表同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 3.10 监控规则表（无需脱敏）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步监控规则表（DWD_DIM_MONITORING_RULE）ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWD.DWD_DIM_MONITORING_RULE PARTITION (etl_date)
select 
    rule_id, 
    rule_name, 
    rule_desc, 
    rule_type, 
    rule_category, 
    active, 
    severity_level, 
    param_json, 
    sql_expression, 
    last_modified_by, 
    last_modified_date,
    '${YESTERDAY}' AS etl_date  -- 动态分区值
from ODS.ODS_MONITORING_RULE
WHERE etl_date = '${YESTERDAY}';  -- 增量过滤
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 监控规则表同步失败！" >> ${LOG_FILE}
    exit 1
fi


echo "[$(date +'%Y-%m-%d %H:%M:%S')] ODS→DWD增量同步完成！" >> ${LOG_FILE}


# 4. DWD→DWS 宽表增量同步（表已存在，仅插入）

echo "[$(date +'%Y-%m-%d %H:%M:%S')] 开始DWD→DWS宽表增量同步ing" >> ${LOG_FILE}

# 4.1 交易分析宽表（DWS_TRANSACTION_ANALYSIS_WIDE）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步交易分析宽表ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE DWS.DWS_TRANSACTION_ANALYSIS_WIDE PARTITION (etl_date)
SELECT
    -- 交易基础维度
    t.txn_type,
    t.txn_sub_type,
    t.txn_channel,
    t.currency,
    -- 时间维度：区分白天/夜间
    CASE WHEN HOUR(t.txn_time) BETWEEN 6 AND 22 THEN '白天' ELSE '夜间' END AS day_night,
    t.txn_status,
    -- 客户维度
    c.cust_type,
    c.occupation,
    c.industry,
    -- 客户收入等级（按年收入划分）
    CASE 
        WHEN c.annual_income >= 100000 THEN '高'
        WHEN c.annual_income >= 50000 THEN '中'
        ELSE '低' 
    END AS income_level,
    c.risk_level,
    -- 账户维度
    a.acct_type,
    a.status AS account_status,
    -- 收款人状态（关联收款人账户的客户状态）
    bc.status AS beneficiary_status,
    -- 日期维度拆分
    t.txn_date AS full_date,
    DAYOFWEEK(t.txn_date) AS day_of_week,
    DAY(t.txn_date) AS day_of_month,
    WEEKOFYEAR(t.txn_date) AS week_of_year,
    MONTHNAME(t.txn_date) AS month_name,
    QUARTER(t.txn_date) AS quarter,
    YEAR(t.txn_date) AS year,
    CASE WHEN DAYOFWEEK(t.txn_date) IN (1,7) THEN '是' ELSE '否' END AS is_weekend,
    -- 财务周期（年+季度）
    CONCAT(YEAR(t.txn_date), 'Q', QUARTER(t.txn_date)) AS fiscal_period,
    -- 交易标识与金额
    t.txn_seq_no,
    t.account_id,
    SUM(t.txn_amount) OVER (PARTITION BY t.account_id, t.txn_date) AS total_amount,  -- 账户当日总交易
    u.ubo_id,  -- 关联受益人ID（替换原简化逻辑）
    t.txn_amount,
    c.customer_id,
    -- 风险维度：当前/历史风险等级
    c.risk_level AS current_risk_level,
    LAG(c.risk_level) OVER (PARTITION BY c.customer_id ORDER BY c.updated_date) AS previous_risk_level,
    a.current_balance,
    -- 报告与预警维度
    s.report_date,
    t.txn_id,
    t.txn_time,
    t.remitter_country,
    t.beneficiary_country,
    t.device_id,
    t.geo_location,
    s.str_id,
    s.report_date AS str_report_date,
    s.report_status,
    al.alert_id,
    al.alert_status,
    al.alert_time,
    al.reviewed_date,
    scr.result_id AS screening_id,
    wl.list_type,
    -- 动态分区值（昨天日期）
    '${YESTERDAY}' AS etl_date
FROM 
    DWD.DWD_FACT_TRANSACTION_DETAIL t
-- 关联客户表（主客户）
LEFT JOIN DWD.DWD_DIM_CUSTOMER_MASTER c 
    ON t.account_id = a.account_id AND c.etl_date = '${YESTERDAY}'
-- 关联账户表
LEFT JOIN DWD.DWD_DIM_ACCOUNT_MASTER a 
    ON t.account_id = a.account_id AND a.etl_date = '${YESTERDAY}'
-- 关联受益人表
LEFT JOIN DWD.DWD_DIM_UBO_INFO u 
    ON c.customer_id = u.customer_id AND u.etl_date = '${YESTERDAY}'
-- 关联收款人客户表（收款人账户→客户）
LEFT JOIN DWD.DWD_DIM_ACCOUNT_MASTER ba 
    ON t.beneficiary_acct_no = ba.acct_no AND ba.etl_date = '${YESTERDAY}'
LEFT JOIN DWD.DWD_DIM_CUSTOMER_MASTER bc 
    ON ba.customer_id = bc.customer_id AND bc.etl_date = '${YESTERDAY}'
-- 关联可疑交易报告表
LEFT JOIN DWD.DWD_FACT_SUSPICIOUS_TXN_REPORT s 
    ON t.customer_id = s.customer_id AND s.etl_date = '${YESTERDAY}'
-- 关联预警表
LEFT JOIN DWD.DWD_FACT_ALTER al 
    ON t.account_id = al.account_id AND al.etl_date = '${YESTERDAY}'
-- 关联筛查结果表
LEFT JOIN DWD.DWD_FACT_SCREENING_RESULT scr 
    ON t.customer_id = scr.target_id AND scr.etl_date = '${YESTERDAY}'
-- 关联名单主表
LEFT JOIN DWD.DWD_DIM_WATCHLIST_MASTER wl 
    ON scr.matched_list_id = wl.list_id AND wl.etl_date = '${YESTERDAY}'
WHERE t.etl_date = '${YESTERDAY}'  -- 仅同步当日增量
GROUP BY
    t.txn_type, t.txn_sub_type, t.txn_channel, t.currency,
    CASE WHEN HOUR(t.txn_time) BETWEEN 6 AND 22 THEN '白天' ELSE '夜间' END,
    t.txn_status, c.cust_type, c.occupation, c.industry,
    CASE WHEN c.annual_income >= 100000 THEN '高' WHEN c.annual_income >= 50000 THEN '中' ELSE '低' END,
    c.risk_level, a.acct_type, a.status, bc.status, t.txn_date,
    DAYOFWEEK(t.txn_date), DAY(t.txn_date), WEEKOFYEAR(t.txn_date),
    MONTHNAME(t.txn_date), QUARTER(t.txn_date), YEAR(t.txn_date),
    CASE WHEN DAYOFWEEK(t.txn_date) IN (1,7) THEN '是' ELSE '否' END,
    CONCAT(YEAR(t.txn_date), 'Q', QUARTER(t.txn_date)),
    t.txn_seq_no, t.account_id, t.txn_amount, c.customer_id,
    c.risk_level, LAG(c.risk_level) OVER (PARTITION BY c.customer_id ORDER BY c.updated_date),
    a.current_balance, s.report_date, t.txn_id, t.txn_time,
    t.remitter_country, t.beneficiary_country, t.device_id, t.geo_location,
    s.str_id, s.report_date, s.report_status, al.alert_id, al.alert_status,
    al.alert_time, al.reviewed_date, scr.result_id, wl.list_type, u.ubo_id;
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] DWS交易分析宽表同步失败！" >> ${LOG_FILE}
    exit 1
fi

echo "[$(date +'%Y-%m-%d %H:%M:%S')] DWD→DWS宽表增量同步完成！" >> ${LOG_FILE}


# 5. DWS→ADS 结果层增量同步（表已存在，仅插入）

echo "[$(date +'%Y-%m-%d %H:%M:%S')] 开始DWS→ADS结果层增量同步ing" >> ${LOG_FILE}

# 5.1 客户风险全景视图（ADS_CUSTOMER_RISK_PANORAMA）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步客户风险全景视图ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE ADS.ADS_CUSTOMER_RISK_PANORAMA PARTITION (etl_date)
SELECT
    customer_id,
    MAX(current_risk_level) AS risk_level,
    SUM(current_balance) AS total_balance,
    COUNT(DISTINCT CASE WHEN full_date BETWEEN DATE_SUB('${YESTERDAY}', 7) AND '${YESTERDAY}' THEN txn_id END) AS last_7d_txn_cnt,
    CASE
        WHEN MAX(current_risk_level) = MAX(previous_risk_level) THEN '持平'
        WHEN (CASE
                WHEN MAX(current_risk_level) = '极高' THEN 4
                WHEN MAX(current_risk_level) = '高' THEN 3
                WHEN MAX(current_risk_level) = '中' THEN 2
                WHEN MAX(current_risk_level) = '低' THEN 1
                ELSE 0
              END) >
             (CASE
                WHEN MAX(previous_risk_level) = '极高' THEN 4
                WHEN MAX(previous_risk_level) = '高' THEN 3
                WHEN MAX(previous_risk_level) = '中' THEN 2
                WHEN MAX(previous_risk_level) = '低' THEN 1
                ELSE 0
              END) THEN '升级'
        ELSE '降级'
    END AS risk_change_flag,
    CASE
        WHEN industry IN ('赌博','虚拟货币') THEN '高危'
        ELSE '常规'
    END AS industry_risk_tag,
    '${YESTERDAY}' AS etl_date
FROM DWS.DWS_TRANSACTION_ANALYSIS_WIDE
WHERE etl_date = '${YESTERDAY}'
GROUP BY customer_id,
    CASE WHEN industry IN ('赌博','虚拟货币') THEN '高危' ELSE '常规' END;
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 客户风险全景视图同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 5.2 交易风险报告（ADS_TRANSACTION_RISK_REPORT）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步交易风险报告ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE ADS.ADS_TRANSACTION_RISK_REPORT PARTITION (etl_date)
SELECT
    '${YESTERDAY}' AS report_date,
    CASE
        WHEN txn_type IN ('现金存入','现金取款') THEN '现金'
        WHEN txn_type LIKE '%跨境%' OR remitter_country != beneficiary_country THEN '跨境'
        ELSE '转账'
    END AS txn_type,
    COUNT(DISTINCT txn_id) AS total_txn_cnt,
    COUNT(DISTINCT CASE WHEN txn_amount > 50000 OR txn_channel = 'ATM' THEN txn_id END) AS high_risk_txn_cnt,
    ROUND(SUM(CASE WHEN remitter_country != beneficiary_country THEN txn_amount ELSE 0 END) /
          NULLIF(SUM(txn_amount), 0) * 100, 2) AS cross_border_ratio,
    ROUND(COUNT(CASE WHEN day_night = '夜间' THEN 1 END) * 100.0 / COUNT(*), 2) AS night_txn_ratio,
    COUNT(DISTINCT CASE
        WHEN device_id IN (
            SELECT device_id
            FROM DWS.DWS_TRANSACTION_ANALYSIS_WIDE
            WHERE etl_date = '${YESTERDAY}'
            GROUP BY device_id
            HAVING COUNT(DISTINCT account_id) > 1
        ) THEN account_id
    END) AS same_device_txn_cnt,
    '${YESTERDAY}' AS etl_date
FROM DWS.DWS_TRANSACTION_ANALYSIS_WIDE
WHERE etl_date = '${YESTERDAY}'
GROUP BY
    CASE
        WHEN txn_type IN ('现金存入','现金取款') THEN '现金'
        WHEN txn_type LIKE '%跨境%' OR remitter_country != beneficiary_country THEN '跨境'
        ELSE '转账'
    END;
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 交易风险报告同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 5.3 监管合规月报（ADS_REGULATORY_COMPLIANCE_MONTHLY）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步监管合规月报ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE ADS.ADS_REGULATORY_COMPLIANCE_MONTHLY PARTITION (etl_date)
SELECT
    COUNT(DISTINCT CASE
        WHEN str_report_date BETWEEN DATE_SUB('${YESTERDAY}', 30) AND '${YESTERDAY}'
        AND report_status = 'SUBMITTED' THEN str_id
    END) AS str_submit_count,
    ROUND(COUNT(DISTINCT CASE WHEN risk_level IN ('高','极高') THEN customer_id END) * 100.0 /
          COUNT(DISTINCT customer_id), 2) AS high_risk_cust_ratio,
    ROUND(AVG(CASE
        WHEN alert_status = 'CONFIRMED' AND alert_time IS NOT NULL AND reviewed_date IS NOT NULL
        THEN (unix_timestamp(reviewed_date) - unix_timestamp(alert_time))/3600
    END), 2) AS avg_process_time,
    ROUND(COUNT(DISTINCT CASE WHEN list_type = 'PEP' THEN screening_id END) * 100.0 /
          NULLIF(COUNT(DISTINCT screening_id), 0), 2) AS pep_hit_rate,
    '${YESTERDAY}' AS etl_date
FROM DWS.DWS_TRANSACTION_ANALYSIS_WIDE
WHERE etl_date = '${YESTERDAY}';
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 监管合规月报同步失败！" >> ${LOG_FILE}
    exit 1
fi


# 5.4 预警规则效能分析（ADS_ALERT_RULE_EFFICIENCY）
echo "[$(date +'%Y-%m-%d %H:%M:%S')] 同步预警规则效能分析ing" >> ${LOG_FILE}
hive -e "
INSERT INTO TABLE ADS.ADS_ALERT_RULE_EFFICIENCY PARTITION (etl_date)
SELECT
    ROUND(COUNT(DISTINCT CASE WHEN alert_status = 'CONFIRMED' THEN alert_id END) * 100.0 /
          NULLIF(COUNT(DISTINCT alert_id), 0), 2) AS precision_rate,
    ROUND(COUNT(DISTINCT CASE WHEN alert_status = 'FALSE_POSITIVE' THEN alert_id END) * 100.0 /
          NULLIF(COUNT(DISTINCT alert_id), 0), 2) AS false_positive_rate,
    ROUND(AVG(CASE
        WHEN alert_time IS NOT NULL AND reviewed_date IS NOT NULL
        THEN (unix_timestamp(reviewed_date) - unix_timestamp(alert_time))/3600
    END), 2) AS avg_process_time,
    ROUND(COUNT(DISTINCT str_id) * 100.0 /
          NULLIF(COUNT(DISTINCT CASE WHEN alert_status = 'CONFIRMED' THEN alert_id END), 0), 2) AS str_conversion_rate,
    '${YESTERDAY}' AS etl_date
FROM DWS.DWS_TRANSACTION_ANALYSIS_WIDE
WHERE etl_date = '${YESTERDAY}';
" >> ${LOG_FILE} 2>&1

if [ $? -ne 0 ]; then
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] 预警规则效能分析同步失败！" >> ${LOG_FILE}
    exit 1
fi

echo "[$(date +'%Y-%m-%d %H:%M:%S')] DWS→ADS结果层增量同步完成！" >> ${LOG_FILE}



# 6. 全链路同步完成

echo "[$(date +'%Y-%m-%d %H:%M:%S')] ODS→DWD→DWS→ADS全链路增量同步成功！" >> ${LOG_FILE}
exit 0

#赋予权限
chmod 755 脚本名;

# 每天00:30运行脚本
30 0 * * * /root/kai_shell/full_layer_sync.sh 2>&1;
#验证
corntab -e;